"""
Case Type   : Compatibility
Case Name   : 导入文件格式：.docx、.pdf格式
Create At   : 2024/9/24
Owner       : @zhanghuan96
Description :
        1.创建数据文件
        2.创建兼容B库
        3.建表
        4.导入.docx格式
        5.查询表
        6.截断表
        7.修改文件扩展名为.pdf
        8.导入.pdf格式
        9.清理环境
Expect      :
        1.创建成功
        2.成功
        3.成功
        4.成功
        5.成功
        6.成功
        7.成功
        8.成功
        9.成功
History     :
"""
import os
import unittest

from testcase.COMPATIBILITY.GRAMMAR.LOAD. \
    Opengauss_Compatibility_Grammar_Load_Common import LoadData
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class LoadData0086(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.common = Common()
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.pri_user = Node('PrimaryDbUser')
        self.db_name = 'db_load0086'
        self.tb_name = 'tb_load0086'
        self.file_path = os.path.join(macro.DB_BACKUP_PATH,
                                      'load_data0086.docx')
        self.file_path01 = os.path.join(macro.DB_BACKUP_PATH,
                                        'load_data0086.pdf')
        self.load_data = LoadData()

    def test_load(self):
        text1 = '----step1:创建数据文件; expect:创建成功----'
        self.log.info(text1)
        echo_cmd = f'''rm -rf {self.file_path};
                    echo "1001+报个+vadfa" >> {self.file_path};
                    echo "1002+发而+bbbvx" >> {self.file_path};
                    echo "1003+报个sfd+vadfa" >> {self.file_path};
                    echo "1004+sdfge+bbbvx" >> {self.file_path};
                    '''
        self.log.info(echo_cmd)
        res = self.common.get_sh_result(self.pri_sh.node, echo_cmd)
        self.log.info(res)
        self.assertEqual('', res, '文件写入失败')

        text2 = '----step2:创建兼容B库; expect:创建成功----'
        self.log.info(text2)
        sql_cmd = (f"drop database if exists {self.db_name};"
                   f"create database {self.db_name} dbcompatibility ='B';")

        self.log.info(sql_cmd)
        sql_res = self.pri_sh.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn('CREATE DATABASE', sql_res, '兼容B库创建失败')

        text = '----step3:建表; expect:建表成功----'
        self.log.info(text)
        sql_cmd1 = (f"drop table if exists {self.tb_name};"
                    f"create table {self.tb_name}(c_01 int primary key,"
                    f"c_02 char(30),c_03 text);")
        self.log.info(sql_cmd1)
        sql_res1 = self.pri_sh.execut_db_sql(sql_cmd1, dbname=self.db_name)
        self.log.info(sql_res1)
        self.assertIn('CREATE TABLE', sql_res1, '查询失败')

        text = '----step4:导入.docx格式; expect：成功----'
        self.log.info(text)
        sql_cmd2 = (f"load data infile '{self.file_path}' into table "
                    f"{self.tb_name} fields terminated by '+' ;")
        self.log.info(sql_cmd2)
        sql_res2 = self.pri_sh.execut_db_sql(sql_cmd2, dbname=self.db_name)
        self.log.info(sql_res2)
        self.assertIn('COPY 4', sql_res2, '执行失败' + text)

        text = '----step5:查询表; expect：成功----'
        self.log.info(text)
        sql_cmd3 = f"select * from {self.tb_name};"
        self.log.info(sql_cmd3)
        sql_res3 = self.pri_sh.execut_db_sql(sql_cmd3, dbname=self.db_name)
        self.log.info(sql_res3)
        rows, _ = self.common.parse_sql_query_res(sql_res3)
        expect = [['1001', '报个', 'vadfa'],
                  ['1002', '发而', 'bbbvx'],
                  ['1003', '报个sfd', 'vadfa'],
                  ['1004', 'sdfge', 'bbbvx']]
        self.assertEqual(rows, expect, '执行失败' + text)

        text = '----step6:截断表; expect：成功----'
        self.log.info(text)
        sql_cmd4 = f"truncate table {self.tb_name};"
        self.log.info(sql_cmd4)
        sql_res4 = self.pri_sh.execut_db_sql(sql_cmd4, dbname=self.db_name)
        self.log.info(sql_res4)
        self.assertIn('TRUNCATE', sql_res4)

        text = '----step7:修改文件扩展名为.pdf; expect：成功----'
        self.log.info(text)
        mv_cmd = f'''mv {self.file_path} {self.file_path01};'''
        self.log.info(mv_cmd)
        res = self.common.get_sh_result(self.pri_sh.node, mv_cmd)
        self.log.info(res)
        self.assertEqual('', res, '文件写入失败')

        text = '----step8:导入.pdf格式; expect：成功----'
        self.log.info(text)
        sql_cmd5 = (f"load data infile '{self.file_path01}' into table "
                    f"{self.tb_name} fields terminated by '+' ;")
        self.log.info(sql_cmd5)
        sql_res5 = self.pri_sh.execut_db_sql(sql_cmd5, dbname=self.db_name)
        self.log.info(sql_res5)
        self.assertIn('COPY 4', sql_res5, '执行失败' + text)

    def tearDown(self):
        text = '----step9:清理环境;expect:成功----'
        self.log.info(text)
        sql_cmd6 = f"drop database if exists {self.db_name};"
        self.log.info(sql_cmd6)
        sql_res6 = self.pri_sh.execut_db_sql(sql_cmd6, dbname='postgres')
        self.log.info(sql_res6)
        rm_cmd = f'''rm -rf {self.file_path};
                     rm -rf {self.file_path01}'''
        self.log.info(rm_cmd)
        res = self.common.get_sh_result(self.pri_sh.node, rm_cmd)
        self.log.info(res)
        self.assertIn('DROP DATABASE', sql_res6)
        self.log.info(f'----{os.path.basename(__file__)} end----')
